/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2013_05_14			*/
/*	File:		MrAdmin.cpp			*/
/********************************************************/

#include "MrAdmin.h"

CMrUdp g_CUdp;
bool g_bDebug = false;
MRADMINCONF_S g_stConf;
CMrTmap<UINT, MRDEVICE_S, MRMAPDIGIT_O> g_CTmap;

int main(int argc, char* argv[])
{
	int iIndex = 0;
	int iReturn = 0;
	UINT uiCount = 0;
	UINT uiTotal = 1;
	CMrThread CMulti;
	CMrThread* pCAccept = NULL;
	char szConf[MR_BUFFER] = "MrAdmin.conf";

	struct option stOptions[] = {
		{"conf", 1, 0, 'c'},
		{"thread", 1, 0, 't'},
		{"debug", 0, 0, 'd'},
		{"help", 0, 0, 'h'},
		{"version", 0, 0, 'v'},
		{0, 0, 0, 0},
	};

	while (true) {
		iReturn = getopt_long(argc, argv, "c:t:dhv", stOptions, &iIndex);
		if (iReturn < 0) {
			break;
		}
		switch (iReturn) {
			case 'c':
				strncpy(szConf, optarg, MR_BUFFER);
				break;
			case 't':
				uiTotal = atoi(optarg);
				break;
			case 'd':
				g_bDebug = true;
				break;
			case 'h':
				MrAdminHelp(argv[0]);
				return MR_SUCCESS;
			case 'v':
				MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
				MR_PRINT("%s Build on %s %s.\n", MR_AD_VERSION, __DATE__, __TIME__);
				return MR_SUCCESS;
			default:
				return MR_SUCCESS;
		}
	}

	signal(SIGSEGV, MrAdminQuit);
	signal(SIGTERM, MrAdminQuit);

	/* config */
	if (true == g_bDebug) {
		MR_DEBUG("Initialize config from %s\n", szConf);
	}
	if (MR_SUCCESS == MrAdminConfig(szConf)) {
		if (true == g_bDebug) {
			MR_DEBUG("Port: %s\n", g_stConf.m_szPort);
			MR_DEBUG("MultiAddr: %s\n", g_stConf.m_szMultiAddr);
			MR_DEBUG("MultiPort: %s\n", g_stConf.m_szMultiPort);
		}
	}
	else {
		return MR_FAILURE;
	}

	/* fork */
	if (true != g_bDebug) {
		MrFork();
	}

	/* multi */
	if (true == g_bDebug) {
		MR_DEBUG("Create multicast thread\n");
	}
	CMulti.MrThreadSetStartRoutine(MrAdminMultiCore);
	if (MR_SUCCESS != CMulti.MrThreadCreate(NULL)) {
		return MR_FAILURE;
	}

	/* accept */
	if (true == g_bDebug) {
		MR_DEBUG("MrAdmin will listen...\n");
	}
	if (MR_SUCCESS == g_CTmap.MrTmapCreateLock()) {
		if (MR_SUCCESS == g_CUdp.MrNetInitRecv(NULL, g_stConf.m_szPort)) {
			if (true == g_bDebug) {
				uiTotal = 1;
			}
			pCAccept = new CMrThread[uiTotal];
			for (uiCount = 0; uiCount < uiTotal; uiCount ++) {
				(pCAccept + uiCount)->MrThreadAttrInit();
				(pCAccept + uiCount)->MrThreadSetStartRoutine(MrAdminAcceptCore);
				(pCAccept + uiCount)->MrThreadCreate(NULL);
			}
			for (uiCount = 0; uiCount < uiTotal; uiCount ++) {
				(pCAccept + uiCount)->MrThreadJoin(NULL);
			}
		}
		else {
			MR_ERROR("Failed to MrNetInitRecv\n");
		}
	}
	else {
		MR_ERROR("Failed to MrTmapCreateLock\n");
	}

	return MR_SUCCESS;
}

void MrAdminHelp(char* pszProc)
{
	MR_PRINT("Usage: %s [options]\n", pszProc);
	MR_PRINT("Valid options are:\n");
	MR_PRINT("[*] -c or --conf:       set configure.\n");
	MR_PRINT("[0] -t or --thread:     set thread total.\n");
	MR_PRINT("[0] -d or --debug:      set debug mode.\n");
	MR_PRINT("[0] -h or --help:       show help.\n");
	MR_PRINT("[0] -v or --version:    show version.\n");
	MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
	MR_PRINT("%s Build on %s %s.\n", MR_AD_VERSION, __DATE__, __TIME__);

	return;
}

void MrAdminQuit(int iSignal)
{
	if (true == g_bDebug) {
		MR_ERROR("MrAdmin is stopped by signal[%d]\n", iSignal);
	}

	exit(MR_FAILURE);
}

int MrAdminConfig(char* pszConf)
{
	CMrDb CDb;
	MRCONFIG_S stConf;
	MRPROKV_S* pstKv = NULL;
	int iReturn = MR_SUCCESS;
	char szSql[MR_PACKET] = { 0 };

	memset(&stConf, '\0', sizeof(MRCONFIG_S));
	if (MR_SUCCESS == MrConfig(pszConf, &stConf)) {
		if (NULL != (pstKv = MrProKvMalloc(MR_BUFFER, MR_PACKET))) {
			if (MR_SUCCESS == CDb.MrDbConnect(stConf.m_szHost, stConf.m_szPort)) {
				sprintf(szSql, MR_AD_CONFIG, stConf.m_szDbase, stConf.m_szTable, stConf.m_szID);
				if (CDb.MrDbSend(MR_DEF_SELT, szSql) > 0) {
					if (MR_SUCCESS == CDb.MrDbRecv(pstKv)) {
						memcpy(g_stConf.m_szPort, (pstKv + 0)->m_puszValue, (pstKv + 0)->m_iLength);
						memcpy(g_stConf.m_szMultiAddr, (pstKv + 1)->m_puszValue, (pstKv + 1)->m_iLength);
						memcpy(g_stConf.m_szMultiPort, (pstKv + 2)->m_puszValue, (pstKv + 2)->m_iLength);
					}
					else {
						MR_ERROR("Failed to MrDbRecv\n");
						iReturn = MR_FAILURE;
					}
				}
				else {
					MR_ERROR("Failed to MrDbSend\n");
					iReturn = MR_FAILURE;
				}
				CDb.MrDbClose();
			}
			else {
				MR_ERROR("Failed to MrDbConnect\n");
				iReturn = MR_FAILURE;
			}
			MrProKvFree(&pstKv, MR_BUFFER);
		}
		else {
			MR_ERROR("Failed to MrProKvMalloc\n");
			iReturn = MR_FAILURE;
		}
	}
	else {
		MR_ERROR("Failed to MrConfig\n");
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

void* MrAdminMultiCore(void* pNULL)
{
	CMrPro CPro;
	CMrAvg CAvg;
	CMrCpu CCpu;
	CMrDsk CDsk;
	CMrIos CIos;
	CMrMem CMem;
	CMrTime CTime;
	CMrMulti CMulti;
	int iSize = 0;
	int iCount = 0;
	int iLength = 0;
	MRPROKV_S stKv;
	MRDEVICE_S stDev;
	CMrCmd* pCCmd = NULL;
	char szValue[MR_VALUE] = { 0 };
	UCHAR uszRecv[MR_ETHERNET - MR_MACIPUDP] = { 0 };
	MRPOLLCMD_S stCmd[] = {{MR_PL_AVG, (void*)&stDev.m_stAvg}, {MR_PL_CPU, (void*)&stDev.m_stCpu}, {MR_PL_DSK, (void*)stDev.m_stDsk}, {MR_PL_IOS, (void*)&stDev.m_stIos}, {MR_PL_MMR, (void*)stDev.m_stMem}};

	if (MR_SUCCESS == CMulti.MrNetInitRecv(g_stConf.m_szMultiAddr, g_stConf.m_szMultiPort)) {
		while ((iLength = CMulti.MrNetRecv(uszRecv, MR_ETHERNET - MR_MACIPUDP)) > 0) {
			if (true == g_bDebug) {
				MR_PRINT("\n");
				CTime.MrTimeOn();
				MR_DEBUG("MrAdminMultiCore start\n");
			}

			/* head */
			CPro.MrProParse(uszRecv, iLength);
			if (true == g_bDebug) {
				CPro.MrProDisplay();
			}

			/* body */
			if (CPro.MrProGetLength() > 0) {
				memset(&stDev, '\0', sizeof(MRDEVICE_S));
				MrProKvInput(&stKv, MR_PRO_1, CPro.MrProGetBody(), CPro.MrProGetLength());
				if (MR_PL_HOST == stKv.m_ucKey) {
					/* ID */
					memcpy(szValue, stKv.m_puszValue, stKv.m_iLength);
					szValue[stKv.m_iLength] = '\0';
					stDev.m_uiID = atoi(szValue);
					if (stDev.m_uiID > 0) {
						/* target */
						iCount = 0;
						while (iCount < MR_PL_TGT) {
							switch (stCmd[iCount].m_ucType & CPro.MrProGetAttr()) {
								case MR_PL_AVG:
									pCCmd = &CAvg;
									iSize = sizeof(MRAVG_S);
									break;
								case MR_PL_CPU:
									pCCmd = &CCpu;
									iSize = sizeof(MRCPU_S);
									break;
								case MR_PL_DSK:
									pCCmd = &CDsk;
									iSize = sizeof(MRDSK_S);
									break;
								case MR_PL_IOS:
									pCCmd = &CIos;
									iSize = sizeof(MRIOS_S);
									break;
								case MR_PL_MMR:
									pCCmd = &CMem;
									iSize = sizeof(MRMMR_S);
									break;
								default:
									iSize = 0;
									pCCmd = NULL;
									break;
							}

							if (NULL != pCCmd) {
								pCCmd->MrCmdResultReset();
								pCCmd->MrCmdResultInput(&uszRecv[MR_PRO_HEAD], CPro.MrProGetLength());
								if (true == g_bDebug) {
									MR_PRINT("\nTarget type: 0x%02x\n", stCmd[iCount].m_ucType);
									pCCmd->MrCmdDisplay();
								}
								memcpy(stCmd[iCount].m_pRemark, pCCmd->MrCmdResultGet(), iSize);
							}

							iCount += 1;
						}

						/* CTmap */
						if (true == g_bDebug) {
							MR_PRINT("\n");
							MR_DEBUG("UPDATE map WHERE ID = %d\n", stDev.m_uiID);
						}
						stDev.m_ulTime = time(NULL);
						g_CTmap.MrTmapUpdate(stDev.m_uiID, stDev);
					}
				}
			}
			else {
				MR_ERROR("Failed to MrProGetLength\n");
			}
			memset(uszRecv,'\0',MR_ETHERNET - MR_MACIPUDP);

			if (true == g_bDebug) {
				MR_DEBUG("MrAdminMultiCore end Use time ");
				CTime.MrTimeOff();
				CTime.MrTimeDisplay();
			}
		}
		CMulti.MrNetClose();
	}
	else {
		MR_ERROR("Failed to MrNetInitRecv\n");
	}

	pthread_exit(NULL);
}

void* MrAdminAcceptCore(void* pNULL)
{
	CMrPro CPro;
	CMrSet CSet;
	CMrUdp CUdp;
	CMrTime CTime;
	int iLength = 0;
	MRDEVICE_S stDev;
	USHORT usCount = 0;
	USHORT usTotal = 0;
	MRPROKV_S stKv[MR_CMD];
	char szID[MR_CMD][MR_VALUE];
	UCHAR ucResult = MR_DEF_SUCC;
	UCHAR uszSend[MR_ETHERNET - MR_MACIPUDP] = { 0 };
	UCHAR uszRecv[MR_ETHERNET - MR_MACIPUDP] = { 0 };

	CUdp.MrNetSetSocket(g_CUdp.MrNetGetSocket());
	while (true) {
		/* request */
		memset(uszRecv, '\0', MR_ETHERNET - MR_MACIPUDP);
		if ((iLength = CUdp.MrNetRecv(uszRecv, MR_ETHERNET - MR_MACIPUDP)) > 0) {
			if (true == g_bDebug) {
				MR_PRINT("\n");
				CTime.MrTimeOn();
				MR_DEBUG("MrAdminAcceptCore[%lu] start MrNetRecv length %d\n", pthread_self(), iLength);
			}

			/* head */
			CPro.MrProReset();
			CPro.MrProParse(uszRecv, iLength);
			if (true == g_bDebug) {
				CPro.MrProDisplay();
			}

			/* body */
			ucResult = MR_DEF_SUCC;
			if (CPro.MrProGetLength() > 0) {
				/* input kv */
				memset(szID, '\0', MR_CMD * MR_VALUE);
				memset(stKv, '\0', sizeof(MRPROKV_S) * MR_CMD);
				MrProKvInput(stKv, CPro.MrProGetTotal(), CPro.MrProGetBody(), CPro.MrProGetLength());
				if (true == g_bDebug) {
					MR_DEBUG("HostID:");
				}
				for (usCount = 0; usCount < CPro.MrProGetTotal(); usCount ++) {
					if (MR_PL_HOST == stKv[usCount].m_ucKey) {
						memcpy(szID[usCount], stKv[usCount].m_puszValue, stKv[usCount].m_iLength);
						if (true == g_bDebug) {
							MR_PRINT(" %s", szID[usCount]);
						}
					}
				}
				if (true == g_bDebug) {
					MR_PRINT("\n");
				}

				if (MR_SUCCESS == CSet.MrSetInit(CPro.MrProGetTotal(), sizeof(MRDEVICE_S))) {
					/* put set */
					usTotal = 0;
					if (true == g_bDebug) {
						MR_DEBUG("Unsort:");
					}
					for (usCount = 0; usCount < CPro.MrProGetTotal(); usCount ++) {
						if (true == g_CTmap.MrMapSelect(atoi(szID[usCount]), stDev, sizeof(MRDEVICE_S))) {
							if (time(NULL) - stDev.m_ulTime < MR_TIMEOUT) {
								if (true == g_bDebug) {
									MR_PRINT(" %d", stDev.m_uiID);
								}
								CSet.MrSetIndexPut(usTotal, &stDev);
								usTotal += 1;
							}
						}
					}
					if (true == g_bDebug) {
						MR_PRINT("\n");
					}

					/* sort set */
					if (CSet.MrSetGetCount() > 0) {
						if (MR_PRO_0 != CPro.MrProGetAttr()) {
							switch (CPro.MrProGetAttr()) {
								case MR_PL_AVG:
									CSet.MrSetCompare(MrAdminCompAvg);
									break;
								case MR_PL_CPU:
									CSet.MrSetCompare(MrAdminCompCpu);
									break;
								case MR_PL_DSK:
									CSet.MrSetCompare(MrAdminCompDsk);
									break;
								case MR_PL_IOS:
									CSet.MrSetCompare(MrAdminCompIos);
									break;
								case MR_PL_MMR:
									CSet.MrSetCompare(MrAdminCompMem);
									break;
								default:
									break;
							}
							if (NULL != CSet.MrSetGetCompare()) {
								CSet.MrSetSort();
							}
							else {
								MR_ERROR("Failed to MrSetGetCompare\n");
								ucResult = MR_DEF_NULL;
							}
						}
						else {
							MR_ERROR("Failed to MrProGetAttr\n");
							ucResult = MR_DEF_ATTR;
						}
					}
					else {
						ucResult = MR_DEF_NULL;
					}

					/* output kv */
					if (MR_DEF_SUCC == ucResult) {
						if (true == g_bDebug) {
							MR_DEBUG("Sorted:");
						}
						memset(szID, '\0', MR_CMD * MR_VALUE);
						for (usCount = 0; usCount < CSet.MrSetGetCount(); usCount ++) {
							sprintf(szID[usCount], "%d", ((MRDEVICE_S*)CSet.MrSetIndexGet(usCount))->m_uiID);
							stKv[usCount].m_puszValue = (UCHAR*)szID[usCount];
							stKv[usCount].m_iLength = strlen(szID[usCount]);
							if (true == g_bDebug) {
								MR_PRINT(" %s", szID[usCount]);
							}
						}
						if (true == g_bDebug) {
							MR_PRINT("\n");
						}
					}

					CSet.MrSetDestroy();
				}
				else {
					MR_ERROR("Failed to MrSetInit\n");
					ucResult = MR_DEF_MALC;
				}
			}
			else {
				if (true == g_bDebug) {
					MR_DEBUG("Active packet response\n");
				}
				ucResult = MR_DEF_LENG;
			}

			/* response */
			memset(uszSend, '\0', MR_ETHERNET - MR_MACIPUDP);
			if (MR_DEF_SUCC != ucResult) {
				iLength = MR_PRO_0;
				usTotal = MR_PRO_0;
			}
			else {
				iLength = MrProKvOutput(stKv, usTotal, &uszSend[MR_PRO_HEAD], MR_ETHERNET - MR_MACIPUDP - MR_PRO_HEAD);
			}
			CPro.MrProPutBody(NULL);
			CPro.MrProPutReal(MR_PRO_0);
			CPro.MrProPutTotal(usTotal);
			CPro.MrProPutLength(iLength);
			CPro.MrProPutResult(ucResult);
			CPro.MrProPutExpand(MR_PRO_N);
			CPro.MrProCreate(uszSend, MR_PRO_HEAD);
			CUdp.MrNetSend(uszSend, MR_PRO_HEAD + iLength);
			if (true == g_bDebug) {
				MR_DEBUG("Response length is %d\n", MR_PRO_HEAD + iLength);
			}

			if (true == g_bDebug) {
				MR_DEBUG("MrAdminAcceptCore[%lu] end Use time ", pthread_self());
				CTime.MrTimeOff();
				CTime.MrTimeDisplay();
			}
		}
		else {
			MR_ERROR("Failed to MrNetRecv\n");
		}
	}

	pthread_exit(NULL);
}

int MrAdminCompAvg(const void* pOne, const void* pTwo)
{
	int iReturn = 0;

	if (((MRDEVICE_S*)pOne)->m_stAvg.m_fMin01 > ((MRDEVICE_S*)pTwo)->m_stAvg.m_fMin01) {
		iReturn = 1;
	}
	else if (((MRDEVICE_S*)pOne)->m_stAvg.m_fMin01 < ((MRDEVICE_S*)pTwo)->m_stAvg.m_fMin01) {
		iReturn = -1;
	}
	else {
	}

	return iReturn;
}

int MrAdminCompCpu(const void* pOne, const void* pTwo)
{
	int iReturn = 0;

	if (((MRDEVICE_S*)pOne)->m_stCpu.m_fInt > ((MRDEVICE_S*)pTwo)->m_stCpu.m_fInt) {
		iReturn = 1;
	}
	else if (((MRDEVICE_S*)pOne)->m_stCpu.m_fInt < ((MRDEVICE_S*)pTwo)->m_stCpu.m_fInt) {
		iReturn = -1;
	}
	else {
	}

	return iReturn;
}

int MrAdminCompDsk(const void* pOne, const void* pTwo)
{
	int iReturn = 0;

	if (((MRDEVICE_S*)pOne)->m_stDsk[0].m_uiUsed > ((MRDEVICE_S*)pTwo)->m_stDsk[0].m_uiUsed) {
		iReturn = 1;
	}
	else if (((MRDEVICE_S*)pOne)->m_stDsk[0].m_uiUsed < ((MRDEVICE_S*)pTwo)->m_stDsk[0].m_uiUsed) {
		iReturn = -1;
	}
	else {
	}

	return iReturn;
}

int MrAdminCompIos(const void* pOne, const void* pTwo)
{
	int iReturn = 0;

	if (((MRDEVICE_S*)pOne)->m_stIos.m_fUtl > ((MRDEVICE_S*)pTwo)->m_stIos.m_fUtl) {
		iReturn = 1;
	}
	else if (((MRDEVICE_S*)pOne)->m_stIos.m_fUtl < ((MRDEVICE_S*)pTwo)->m_stIos.m_fUtl) {
		iReturn = -1;
	}
	else {
	}

	return iReturn;
}

int MrAdminCompMem(const void* pOne, const void* pTwo)
{
	int iReturn = 0;

	if (((MRDEVICE_S*)pOne)->m_stMem[0].m_uiFree > ((MRDEVICE_S*)pTwo)->m_stMem[0].m_uiFree) {
		iReturn = 1;
	}
	else if (((MRDEVICE_S*)pOne)->m_stMem[0].m_uiFree < ((MRDEVICE_S*)pTwo)->m_stMem[0].m_uiFree) {
		iReturn = -1;
	}
	else {
	}

	return iReturn;
}
